package datastream.api.operator.multi;

import datastream.api.operator.multi.pojo.log.InputModel1;
import datastream.api.operator.multi.pojo.log.InputModel2;
import datastream.api.operator.multi.pojo.log.OutputModel;
import datastream.api.operator.multi.source.log.InputModel1Source;
import datastream.api.operator.multi.source.log.InputModel2Source;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

public class CoFlatMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<InputModel1> source1 = env.addSource(new InputModel1Source());
        DataStreamSource<InputModel2> source2 = env.addSource(new InputModel2Source());

        ConnectedStreams<InputModel1, InputModel2> connect = source1.connect(source2);

        SingleOutputStreamOperator<OutputModel> coFlatMap = connect.flatMap(new CoFlatMapFunction<InputModel1, InputModel2, OutputModel>() {
            @Override
            public void flatMap1(InputModel1 value, Collector<OutputModel> out) throws Exception {

                for (Tuple2<String, String> log : value.getBatchLog()) {
                    out.collect(OutputModel.builder()
                            .username(value.getUsername())
                            .log(log)
                            .build());
                }
            }

            @Override
            public void flatMap2(InputModel2 value, Collector<OutputModel> out) throws Exception {
                out.collect(OutputModel.builder()
                        .username(value.getUsername())
                        .log(value.getLog())
                        .build());
            }
        });

        DataStreamSink<OutputModel> sink = coFlatMap.print();

        env.execute();
    }
}
